/*************************************************************************
    > File Name    : EventLoop.cc
    > Author       : LiuLei
    > Mail         : 2261459486@qq.com 
    > Created Time : 2022年11月25日 星期五 13时04分28秒
 ************************************************************************/

#include "../../include/server/EventLoop.h"
#include "../../include/server/Acceptor.h"
#include "../../include/server/MutexLockGuard.h"
#include "../../include/server/TcpConnection.h"
#include <unistd.h>
#include <iostream>
using std::cout;
using std::endl;

EventLoop::EventLoop(Acceptor &acceptor)
: _epfd(createEpollFd())
, _evfd(createEventFd())
, _timerfd(createTimerFd())
, _isLooping(false)
, _evtList(1024)
, _conns()
, _pendings()
, _acceptor(acceptor)
, _mutex()
, _roundRobin(30)
{
    //监听listenfd
    this->addEpollReadFd(_acceptor.getSocketFd());
    //监听eventfd的返回值
    this->addEpollReadFd(_evfd);
    //不在这里监听，在loop函数中监听
    /* //监听timerfd的返回值 */
    /* addEpollReadFd(_timerfd); */
}

EventLoop::~EventLoop()
{
    if(_epfd >= 0)
    {
        ::close(_epfd);
    }
    if(_evfd >= 0)
    {
        ::close(_evfd);
    }
    if(_timerfd >= 0)
    {
        //停止计时器
        this->setTimerFd(0, 0);
        ::close(_timerfd);
    }
}

void EventLoop::loop()
{
    _isLooping = true;

    cout << "初始化已完成！" << endl;

    this->setTimerFd(1, 1);
    //监听timerfd的返回值
    this->addEpollReadFd(_timerfd);

    while(_isLooping)
    {
        this->waitEpollFd();
    }
}

void EventLoop::unloop()
{
    _isLooping = false;
}

void EventLoop::setConnectionCallback(TcpConnectionCallback &&cb)
{
    _onConnection = std::move(cb);
}

void EventLoop::setMessageCallback(TcpConnectionCallback &&cb)
{
    _onMessage = std::move(cb);
}

void EventLoop::setCloseCallback(TcpConnectionCallback &&cb)
{
    _onClose = std::move(cb);
}

void EventLoop::runInLoop(Function &&cb)
{
    {
        MutexLockGuard autoLock(_mutex);
        _pendings.push_back(std::move(cb));
    }
    this->weakup();
}

void EventLoop::waitEpollFd()
{
    int nready;
    do
    {
        //因为epoll_wait函数的第二个参数是struct epoll_event *                
        //然而_evtList 的类型是 vector<struct epoll_event>                    
        //先获取迭代器，再解引用再取地址                                      
        /* int nready = epoll_wait(_epfd, _evtList, _evtList.size(), 5000); */
        nready = ::epoll_wait(_epfd, &*_evtList.begin(), _evtList.size(), 5000);
    } while(-1 == nready && errno == EINTR);

    if(-1 == nready)
    {
        ::perror("epoll_wait in EventLoop::waitEpollFd()");
    }
    else if(0 == nready)
    {
        cout << ">> epoll_wait timeout!" << endl;
    }
    else
    {
        //考虑要不要扩容
        if(nready == static_cast<int>(_evtList.size()))
        {
            _evtList.resize(2 * nready);
        }
        
        for(int idx = 0; idx < nready; ++idx)
        {
            int fd = _evtList[idx].data.fd;
            //新的连接
            if(fd == _acceptor.getSocketFd())
            {
                //读事件
                if(_evtList[idx].events & EPOLLIN)
                {
                    this->handleNewConnection();
                }
            }
            //监听的_evfd有事件发生
            else if(fd == _evfd)
            {
                if(_evtList[idx].events & EPOLLIN)
                {
                    this->handleRead();             //阻塞

                    this->doPendingFunctions();     //执行所有任务
                }
            }
            //监听的_timerfd有事件发生
            else if(fd == _timerfd)
            {
                if(_evtList[idx].events & EPOLLIN)
                {
                    this->handleTimerFd();
                }
            }
            //老的连接
            else
            {
                if(_evtList[idx].events & EPOLLIN)
                {
                    this->handleMessage(fd);
                }
            }
        }
    }
}

void EventLoop::handleNewConnection()
{
    int peerfd = _acceptor.accpet();
    if(peerfd < 0)
    {
        ::perror("error in EventLoop::handleNewConnection()");
    }

    this->addEpollReadFd(peerfd);

    TcpConnectionPtr con(new TcpConnection(peerfd, this));

    //三个事件注册
    con->setConnectionCallback(_onConnection);
    con->setMessageCallback(_onMessage);
    con->setCloseCallback(_onClose);

    //插入
    _conns.insert(std::make_pair(peerfd, con));

    //注册事件执行
    con->handleConnectionCallback();
}

void EventLoop::handleMessage(int fd)
{
    auto it = _conns.find(fd);
    if(it != _conns.end())
    {
        //直接判断读的结果是否为0
        bool flag = it->second->isClosed();
        if(flag)
        {
            //处理连接的断开
            it->second->handleCloseCallback();
            this->deleteEpollReadFd(fd);
            _roundRobin.removeOneConnection(fd);    //轮询也要删除
            _conns.erase(it);
        }
        else
        {
            //连接正常，执行操作
            _roundRobin.updateOneConnection(fd);    //轮询也要更新
            it->second->handleMessageCallback();
        }
    }
    else
    {
        cout << "该连接不存在！" << endl;
    }
}

int EventLoop::createEpollFd()
{
    int fd = ::epoll_create1(0);
    if(fd < 0)
    {
        ::perror("epoll_create1 in EventLoop::createEpollFd()");
    }
    return fd;
}

void EventLoop::addEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN;
    int ret = ::epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &evt);
    if(ret < 0)
    {
        ::perror("epoll_ctl in EventLoop::addEpollReadFd()");
    }
}

void EventLoop::deleteEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    int ret = ::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, &evt);
    if(ret < 0)
    {
        ::perror("epoll_ctl in EventLoop::deleteEpollReadFd()");
    }
}

int EventLoop::createEventFd()
{
    int fd = ::eventfd(0, 0);
    if(fd < 0)
    {
        ::perror("eventfd in EventLoop::createEventFd()");
    }
    return fd;
}

int EventLoop::createTimerFd()
{
    int fd = ::timerfd_create(CLOCK_REALTIME, 0);
    if(fd < 0)
    {
        ::perror("timerfd_create in EventLoop::createTimerFd()");
        return fd;
    }
    return fd;
}

void EventLoop::setTimerFd(int initSec, int peridocSec)
{
    struct itimerspec value;
    value.it_value.tv_sec = initSec;                            //起始时间
    value.it_value.tv_nsec = 0;
    
    value.it_interval.tv_sec = peridocSec;                      //周期时间
    value.it_interval.tv_nsec = 0;
                                                            
    int ret = timerfd_settime(_timerfd, 0, &value, nullptr);
    if(ret < 0)
    {
        perror("timerfd_settime in EventLoop::setTimerFd()");
        return ;
    }
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    int ret = ::read(_evfd, &one, sizeof(one));
    if(ret != sizeof(one))
    {
        ::perror("read in EventLoop::handleRead()");
        return ;
    }
}

void EventLoop::handleTimerFd()
{
    uint64_t one = 1;
    int ret = ::read(_evfd, &one, sizeof(one));
    if(ret != sizeof(one))
    {
        ::perror("read in EventLoop::handleTimerFd()");
        return ;
    }
    
    //查询过期的连接
    unordered_set<int> delConn = _roundRobin.autoUpdate();
    for(auto &fd : delConn)
    {
        this->deleteEpollReadFd(fd);
        cout << _conns[fd]->addrString() << " time out!" << endl;
        _conns.erase(fd);
    }
}

void EventLoop::weakup()
{
    uint64_t one = 1;
    int ret = ::write(_evfd, &one, sizeof(one));
    if(ret != sizeof(one))
    {
        ::perror("write in EventLoop::weakup()");
        return ;
    }
}

void EventLoop::doPendingFunctions()
{
    vector<Function> tmp;
    {
        MutexLockGuard autoLock(_mutex);
        tmp.swap(_pendings);
    }

    for(auto &cb : tmp)
    {
        cb();
    }
}
